package day06

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 第五章 广播变量
 *
 * 广播变量：分布式共享只读变量
 *
 * 在多个并行操作中(Executor)使用同一个变量，Spark默认会为每个任务(Task)分别发送，这样
 * 如果共享比较大的对象，会占用很大工作节点的内存。
 *
 * 广播变量用来高效分发较大的对象。向所有工作借点发送一个较大的只读值，以供一个或多个Spark
 * 操作使用。比如，如果你的应用需要向所有节点发送一个较大的只读查询表，甚至是机器学习算法中的
 * 很大的特征向量，广播变量用起来都很顺手。
 *
 * (1)使用广播变量的步骤：
 * ① 通过对一个类型T的对象调用SparkContext.broadcast创建出一个Broadcast[T]对象，任何
 * 可序列化的类型都可以这么实现
 * ② 通过value属性访问该对象的值
 * ③ 变量只会被发到各个节点一次，应作为只读值处理(修改这个值不会影响到别的节点)。
 *
 * (2)原理说明：
 *
 */
object BroadcastVar {

  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.创建RDD
    val rdd1: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 3)))
    val rdd2: RDD[(String, Int)] = sc.makeRDD(List(("a", 4), ("b", 5), ("c", 6)))

    //3.1 采用RDD的方式实现 rdd1 join rdd2，用到Shuffle，性能比较低?待验证
    val joinRdd: RDD[(String, (Int, Int))] = rdd1.join(rdd2)
    joinRdd.foreach(println)

    println("--------------------------------------------")

    // 声明广播变量
    val rdd3: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 3)))
    val list: List[(String, Int)] = List(("a", 4), ("b", 5), ("c", 6))

    val broadcastList: Broadcast[List[(String, Int)]] = sc.broadcast(list)

    val resRdd: RDD[(String, (Int, Int))] = rdd3.map {
      case (k1, v1) =>
        var v2: Int = 0
        for ((k3, v3) <- broadcastList.value) {
          if (k3 == k1) {
            v2 = v3
          }
        }

        (k1, (v1, v2))
    }

    resRdd.foreach(println)

    // 关闭连接
    sc.stop()
  }

}
